/*
 * Copyright 2011-2013 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.udream.support.redis;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.data.redis.connection.jedis.JedisUtils;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.Protocol;

/**
 * Connection factory creating
 * <a href="http://github.com/xetorthio/jedis">Jedis</a> based connections.
 * 
 * @author Costin Leau
 */
public class JedisConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory, JedisSupport {

 private final static Log log = LogFactory.getLog(JedisConnectionFactory.class);

 private JedisShardInfo shardInfo;
 private String hostName = "localhost";
 private int port = Protocol.DEFAULT_PORT;
 private int timeout = Protocol.DEFAULT_TIMEOUT;
 private String password;

 private boolean usePool = true;
 private JedisPool pool = null;
 private JedisPoolConfig poolConfig = new JedisPoolConfig();

 private int dbIndex = 0;

 /**
  * Constructs a new <code>JedisConnectionFactory</code> instance with default
  * settings (default connection pooling, no shard information).
  */
 public JedisConnectionFactory() {
 }

 /**
  * Constructs a new <code>JedisConnectionFactory</code> instance. Will override
  * the other connection parameters passed to the factory.
  *
  * @param shardInfo
  *         shard information
  */
 public JedisConnectionFactory(JedisShardInfo shardInfo) {
  this.shardInfo = shardInfo;
 }

 /**
  * Constructs a new <code>JedisConnectionFactory</code> instance using the
  * given pool configuration.
  *
  * @param poolConfig
  *         pool configuration
  */
 public JedisConnectionFactory(JedisPoolConfig poolConfig) {
  this.poolConfig = poolConfig;
 }

 /**
  * Returns a Jedis instance to be used as a Redis connection. The instance can
  * be newly created or retrieved from a pool.
  * 
  * @return Jedis instance ready for wrapping into a {@link RedisConnection}.
  */
 protected Jedis fetchJedisConnector() {
  try {
   if (usePool && pool != null) {
    return pool.getResource();
   }
   Jedis jedis = new Jedis(getShardInfo());
   // force initialization (see Jedis issue #82)
   jedis.connect();
   return jedis;
  } catch (Exception ex) {
   throw new RedisConnectionFailureException("Cannot get Jedis connection", ex);
  }
 }

 /**
  * Post process a newly retrieved connection. Useful for decorating or
  * executing initialization commands on a new connection. This implementation
  * simply returns the connection.
  * 
  * @param connection
  * @return processed connection
  */
 protected JedisConnection postProcessConnection(JedisConnection connection) {
  return connection;
 }

 public void afterPropertiesSet() {
  if (shardInfo == null) {
   shardInfo = new JedisShardInfo(hostName, port);

   if (StringUtils.hasLength(password)) {
    shardInfo.setPassword(password);
   }

   if (timeout > 0) {
    shardInfo.setTimeout(timeout);
   }
  }

  if (usePool) {
   pool = new JedisPool(poolConfig, shardInfo.getHost(), shardInfo.getPort(), shardInfo.getTimeout(), shardInfo.getPassword());
  }
 }

 public void destroy() {
  if (usePool && pool != null) {
   try {
    pool.destroy();
   } catch (Exception ex) {
    log.warn("Cannot properly close Jedis pool", ex);
   }
   pool = null;
  }
 }

 public JedisConnection getConnection() {
  Jedis jedis = fetchJedisConnector();
  return postProcessConnection((usePool ? new JedisConnection(jedis, pool, dbIndex) : new JedisConnection(jedis, null, dbIndex)));
 }

 public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
  return JedisUtils.convertJedisAccessException(ex);
 }

 /**
  * Returns the Redis hostName.
  *
  * @return Returns the hostName
  */
 public String getHostName() {
  return hostName;
 }

 /**
  * Sets the Redis hostName.
  * 
  * @param hostName
  *         The hostName to set.
  */
 public void setHostName(String hostName) {
  this.hostName = hostName;
 }

 /**
  * Returns the password used for authenticating with the Redis server.
  * 
  * @return password for authentication
  */
 public String getPassword() {
  return password;
 }

 /**
  * Sets the password used for authenticating with the Redis server.
  * 
  * @param password
  *         the password to set
  */
 public void setPassword(String password) {
  this.password = password;
 }

 /**
  * Returns the port used to connect to the Redis instance.
  * 
  * @return Redis port.
  */
 public int getPort() {
  return port;

 }

 /**
  * Sets the port used to connect to the Redis instance.
  * 
  * @param port
  *         Redis port
  */
 public void setPort(int port) {
  this.port = port;
 }

 /**
  * Returns the shardInfo.
  *
  * @return Returns the shardInfo
  */
 public JedisShardInfo getShardInfo() {
  return shardInfo;
 }

 /**
  * Sets the shard info for this factory.
  * 
  * @param shardInfo
  *         The shardInfo to set.
  */
 public void setShardInfo(JedisShardInfo shardInfo) {
  this.shardInfo = shardInfo;
 }

 /**
  * Returns the timeout.
  *
  * @return Returns the timeout
  */
 public int getTimeout() {
  return timeout;
 }

 /**
  * @param timeout
  *         The timeout to set.
  */
 public void setTimeout(int timeout) {
  this.timeout = timeout;
 }

 /**
  * Indicates the use of a connection pool.
  *
  * @return Returns the use of connection pooling.
  */
 public boolean getUsePool() {
  return usePool;
 }

 /**
  * Turns on or off the use of connection pooling.
  * 
  * @param usePool
  *         The usePool to set.
  */
 public void setUsePool(boolean usePool) {
  this.usePool = usePool;
 }

 /**
  * Returns the poolConfig.
  *
  * @return Returns the poolConfig
  */
 public JedisPoolConfig getPoolConfig() {
  return poolConfig;
 }

 /**
  * Sets the pool configuration for this factory.
  * 
  * @param poolConfig
  *         The poolConfig to set.
  */
 public void setPoolConfig(JedisPoolConfig poolConfig) {
  this.poolConfig = poolConfig;
 }

 /**
  * Returns the index of the database.
  *
  * @return Returns the database index
  */
 public int getDatabase() {
  return dbIndex;
 }

 /**
  * Sets the index of the database used by this connection factory. Default is
  * 0.
  * 
  * @param index
  *         database index
  */
 public void setDatabase(int index) {
  Assert.isTrue(index >= 0, "invalid DB index (a positive index required)");
  this.dbIndex = index;
 }

 @Override
 public Jedis getJedis() {
  return this.fetchJedisConnector();
 }

 @Override
 public void returnJedis(Jedis jedis) {
  if (usePool && pool != null) {
   pool.returnResource(jedis);
  } else {
   if (jedis.isConnected()) {
    jedis.disconnect();
   }
  }
 }

 @Override
 public Object doInJedis(JedisCallback callback) {
  Jedis jedis = this.getJedis();
  try {
   Object result = callback.doWithJedis(jedis);
   return result;
  } finally {
   returnJedis(jedis);
  }
 }

}